package com.nit.netty.zookeeper.registry;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.springframework.stereotype.Component;

import java.util.concurrent.CountDownLatch;

/**
 * @Author: whb
 * @Description:
 * @CreateTime: 2020/8/24 16:11
 */
@Component
@Slf4j
public class ServiceRegistry implements Watcher {
    private ZooKeeper zk;
    private static final int SESSION_TIMEOUT = 15000;
    private static final CountDownLatch countDownLatch = new CountDownLatch(1);
    private String zkAddress;
    public ServiceRegistry() {

    }

    private Stat exists(String node, boolean needWatch) {
        try {
            return zk.exists(node, needWatch);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    public ServiceRegistry(String zkAddress) {
        connect(zkAddress);
    }

    private void connect(String zkAddress) {
        try {
            zk = new ZooKeeper(zkAddress, SESSION_TIMEOUT, this);
            countDownLatch.await();
            this.zkAddress = zkAddress;
            log.debug("zookeeper连接成功");
        } catch (Exception ex) {
            log.error("连接zookeeper失败", ex);
        }
    }

    private static final String REGISTRY_PATH = "/registry";

    public void register(String serviceName, String serviceAddress) {
        try {
            String registryPath = REGISTRY_PATH;
            if (exists(registryPath, false) == null) {
                zk.create(registryPath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                log.debug("创建了节点:{}", registryPath);
            }
            //创建服务节点（持久节点）
            String servicePath = registryPath + "/" + serviceName;
            if (exists(servicePath, false) == null) {
                zk.create(servicePath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                log.debug("创建了节点:{}", servicePath);
            }
            //创建地址节点
            String addressPath = servicePath + "/address-";
            String addressNode = zk.create(addressPath, serviceAddress.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            log.debug("创建了节点:{} => {}", addressNode, serviceAddress);
        } catch (Exception e) {
            log.error("创建节点失败", e);
        }
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        // 获取事件状态
        Event.KeeperState keeperState = watchedEvent.getState();
        // 获取事件类型
        Event.EventType eventType = watchedEvent.getType();
        // 判断是否建立连接
        if (Event.KeeperState.SyncConnected == keeperState) {
            // 如果当前状态已经连接上了 SyncConnected：连接，AuthFailed：认证失败,Expired:失效过期,
            // ConnectedReadOnly:连接只读,Disconnected:连接失败
            if (Event.EventType.None == eventType) {
                // 如果建立建立成功,让后程序往下走
                log.info("zk 建立连接成功!");
                countDownLatch.countDown();
            }
        } else if (Event.KeeperState.Expired == keeperState) {
            connect(zkAddress);
        }
    }
}
